use crate::imp::tonicrpc::ekhokcp::kcp_imp::is_steam_backpressure_heavy;
use crate::imp::tonicrpc::kcptunnel::tunnel::{
    IceAuth, IceCandidate, IceCommandMsg, IceCreateMsg, TonicServerStopKind, TunnelClientCreator,
    TunnelConfig, TunnelConn, TunnelConnectMsg, TunnelCreatorImp, TunnelPair, TunnelSignalIceMsg,
    TunnelStatus, TunnelStopKind,
};
use crate::imp::tonicrpc::signal::SignalAdapter;
use crate::imp::tonicrpc::signal::TunnelSignalTrait;
use crate::imp::tonicrpc::tonicserver::{
    close_remote_tonic_tunnel, get_or_set_bytebuffer_sender, get_or_set_content_sender,
    if_no_kcp_set_err, FlowControlInterceptor, SendByteBufferCommand, SendContentCommand,
};
use crate::imp::tonicrpc::tonicserverimp::set_download_base_path;

use crate::utils::ids;
use crate::utils::query;
use async_trait::async_trait;
// use tokio::{
//     fs::{remove_file, rename, File, OpenOptions},
//     io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt, BufReader, BufWriter},
// };
use tokio::{runtime::Builder, sync::RwLock, time::Instant};

use webrtc_ice::state::ConnectionState;

use std::collections::{HashMap, VecDeque};
use std::mem;
use std::sync::Arc;
use std::time::Duration;

use futures_channel::mpsc::{self, UnboundedSender};
use futures_util::{
    future::{self, Either},
    pin_mut, AsyncReadExt, AsyncWriteExt, SinkExt, StreamExt,
};

use crate::imp::ws::websocket_imp::MessageToReceived;
use crate::imp::ws::websocket_imp::WsAckSendResult;
use crate::imp::ws::websocket_imp::WsReceivedMessage;
use crate::utils::time_utils;
use tonic::{codegen::InterceptedService, transport::Channel, Request};
use tonic_inf::api::{
    byte_buffer_msg::{
        self,
        Bfmsg::{self, BfPing, BfPong, BfRes, BfStart, BfStop},
    },
    send_content_msg::{
        self,
        Msg::{self, Ping, Pong, Res, Start, Stop},
    },
    tunnel_service_client::TunnelServiceClient,
    ByteBufferData, ByteBufferMsg, ByteBufferPing, ByteBufferPong, ByteBufferRes, ByteBufferStart,
    ByteBufferStop, SendContentData, SendContentMsg, SendContentPing, SendContentPong,
    SendContentRes, SendContentStart, SendContentStop,
};

const TUNNEL_EVENT: &'static str = "__obrr_tunnel_event__";

pub trait TonicTunnelCallback: Send + Sync {
    // tonic server created
    fn on_tonic_server_created(&self, creator: TunnelCreatorImp);
    // tonic server droped
    fn on_tonic_server_droped(&self, msg: TonicServerStopKind);
    // kcp tunnel ready
    fn on_tunnel_connect_ready(&self, msg: TunnelConn);
    // kcp tunnel destroy
    fn on_tunnel_connect_destroy(
        &self,
        tunnel_id: String,
        local_client_id: String,
        remote_client_id: String,
    );
    // kcp tunnel state changed
    fn on_tunnel_state_changed(&self, msg: TunnelStatus);
    // kcp tunnel pair changged
    fn on_tunnel_pair_changed(&self, msg: TunnelPair);
    // received remote send content
    fn on_remote_content(&self, data: SendContentData) -> i64;
    // received send content result
    fn on_send_content_result(&self, res: SendContentRes) -> i64;
    // received tonic content service start
    fn on_content_service_start(&self, info: SendContentStart) -> i64;
    // received tonic content service stop
    fn on_content_service_stop(&self, tunnel_id: String) -> i64;

    // received remote send bytebuffer
    fn on_remote_bufferbuffer(&self, data: ByteBufferData) -> i64;
    // received send bytebuffer result
    fn on_send_bytebuffer_result(&self, res: ByteBufferRes) -> i64;
    // received tonic bytebuffer service start
    fn on_bytebuffer_service_start(&self, info: ByteBufferStart) -> i64;
    // received tonic bytebuffer service stop
    fn on_bytebuffer_service_stop(&self, tunnel_id: String) -> i64;
}

async fn disconnect(
    map: &Arc<
        RwLock<HashMap<String, (UnboundedSender<TunnelSignalIceMsg>, String, ConnectionState)>>,
    >,
    sender_tx: &mut UnboundedSender<(String, TunnelSignalIceMsg)>,
    my_device_id: &str,
    tunnel_id: &str,
) -> Result<(), anyhow::Error> {
    log::info!("disconnect tunnel_id: {}", tunnel_id);
    let channel = {
        let mut map = map.write().await;
        map.remove(tunnel_id)
    };

    match channel {
        Some((mut sender, remote_device_id, _)) => {
            if let Err(e) = sender_tx
                .send((
                    remote_device_id.clone(),
                    TunnelSignalIceMsg::Stop {
                        tunnel_id: tunnel_id.to_owned(),
                        st_local_client_id: my_device_id.to_owned(),
                        st_remote_client_id: remote_device_id.to_owned(),
                        kind: TunnelStopKind::RemoteClose,
                    },
                ))
                .await
            {
                log::error!("disconnect  send err: {}", &e);
            }

            sender
                .send(TunnelSignalIceMsg::Stop {
                    tunnel_id: tunnel_id.to_owned(),
                    st_local_client_id: my_device_id.to_owned(),
                    st_remote_client_id: remote_device_id.to_owned(),
                    kind: TunnelStopKind::LocalClose,
                })
                .await?;
        }
        None => {
            // log::error!("disconnect tunnel_id {} not found", tunnel_id);
            anyhow::bail!("disconnect tunnel_id {} not found", tunnel_id);
        }
    }
    Ok(())
}

fn get_tunnel_id_from_msg(msg: TunnelSignalIceMsg) -> (String, String) {
    let (tunnel_id, device_id) = match msg {
        TunnelSignalIceMsg::RemoteAuth(IceAuth {
            tunnel_id,
            auth_local_client_id: _,
            auth_remote_client_id: device_id,
            ..
        }) => (tunnel_id, device_id),
        TunnelSignalIceMsg::RemoteCandidate(IceCandidate {
            tunnel_id,
            can_local_client_id: _,
            can_remote_client_id: device_id,
            ..
        }) => (tunnel_id, device_id),
        TunnelSignalIceMsg::Stop {
            tunnel_id,
            st_local_client_id: _,
            st_remote_client_id: device_id,
            ..
        } => (tunnel_id, device_id),
        TunnelSignalIceMsg::GatherCandidates {
            tunnel_id,
            gc_local_client_id: _,
            gc_remote_client_id: device_id,
            ..
        } => (tunnel_id, device_id),
    };

    log::info!(
        "get_tunnel_id_from_msg tunnel_id: {} device_id: {}",
        tunnel_id,
        device_id
    );
    (tunnel_id, device_id)
}

fn get_tunnel_id_from_data(data: &str) -> Result<(String, String), anyhow::Error> {
    let msg = serde_json::from_str::<TunnelSignalIceMsg>(data)?;
    let (tunnel_id, device_id) = match msg {
        TunnelSignalIceMsg::RemoteAuth(IceAuth {
            tunnel_id,
            auth_local_client_id: _,
            auth_remote_client_id: device_id,
            ..
        }) => (tunnel_id, device_id),
        TunnelSignalIceMsg::RemoteCandidate(IceCandidate {
            tunnel_id,
            can_local_client_id: _,
            can_remote_client_id: device_id,
            ..
        }) => (tunnel_id, device_id),
        TunnelSignalIceMsg::Stop {
            tunnel_id,
            st_local_client_id: _,
            st_remote_client_id: device_id,
            ..
        } => (tunnel_id, device_id),
        TunnelSignalIceMsg::GatherCandidates {
            tunnel_id,
            gc_local_client_id: _,
            gc_remote_client_id: device_id,
            ..
        } => (tunnel_id, device_id),
    };
    log::info!(
        "get_tunnel_id_from_data tunnel_id: {} device_id: {}",
        tunnel_id,
        device_id
    );
    Ok((tunnel_id, device_id))
}

#[async_trait]
impl TunnelClientCreator for TunnelCreatorImp {
    async fn request_connect(
        &mut self,
        remote_client_id: &str,
        tunnel_id: &str,
    ) -> Result<String, anyhow::Error> {
        // create a tunnel id,
        let tunnel_id = if tunnel_id == "" {
            ids::id_generate_xbit(6)
        } else {
            tunnel_id.to_owned()
        };

        // insert recieve_rx to signal_channel_map.
        let (receive_tx, receive_rx) = mpsc::unbounded::<TunnelSignalIceMsg>();
        {
            // insert signal channel map.
            self.signal_channel_map.write().await.insert(
                tunnel_id.to_owned(),
                (
                    receive_tx,
                    remote_client_id.to_owned(),
                    ConnectionState::Unspecified,
                ),
            );
        };

        // self commander send create message.
        self.commander
            .send(IceCommandMsg::Create(IceCreateMsg {
                is_controlling: true,
                tunnel_id: tunnel_id.to_owned(),
                remote_client_id: remote_client_id.to_owned(),
                receive_rx,
            }))
            .await?;

        Ok(tunnel_id)
    }

    fn disconnect(&mut self, tunnel_id: &str) -> Result<(), anyhow::Error> {
        log::info!("disconnect tunnel_id: {}", tunnel_id);
        self.commander
            .unbounded_send(IceCommandMsg::DisconnectTunnel(tunnel_id.to_owned()))?;
        Ok(())
    }

    fn gather_candidates(&mut self) -> Result<(), anyhow::Error> {
        log::info!("gather_candidates");
        self.commander
            .unbounded_send(IceCommandMsg::GatherCandidates)?;
        Ok(())
    }

    fn stop_tonic_server(&mut self) -> Result<(), anyhow::Error> {
        if let Err(e) = self.commander.unbounded_send(IceCommandMsg::StopService) {
            log::error!("stop_service commander.send err: {}", &e);
            anyhow::bail!(e);
        }
        Ok(())
    }
}

impl TunnelCreatorImp {
    // send disconnect command by channel.
    // see start_tonic_tunnel::command_handler.
    pub fn disconnect(&mut self, tunnel_id: &str) -> Result<(), anyhow::Error> {
        log::info!("disconnect tunnel_id: {}", tunnel_id);
        self.commander
            .unbounded_send(IceCommandMsg::DisconnectTunnel(tunnel_id.to_owned()))?;
        Ok(())
    }

    pub fn stop_tonic_server(&self) -> Result<(), anyhow::Error> {
        if let Err(e) = self.commander.unbounded_send(IceCommandMsg::StopService) {
            log::error!("stop_service commander.send err: {}", &e);
            anyhow::bail!(e);
        }
        Ok(())
    }

    //async_new_tunnel_connect implementation in tunnel.rs
    pub async fn start_tonic_tunnel(
        client_id: &str,
        mut callback_sender: UnboundedSender<TunnelConnectMsg>,
        options: &TunnelConfig,
    ) -> Result<(), anyhow::Error> {
        // set the download base path.
        set_download_base_path(options.download_base_path.clone()).await;

        // create a signal adapter with tunnel_event
        let (signal_adapter, mut signal_receiver) =
            SignalAdapter::create(vec![TUNNEL_EVENT.to_owned()]).await?;

        let mut signal_adapter_clone = signal_adapter.clone();
        // create signal hashmap,key = tunnel_id, value = tuple((UnboundedSender<TunnelSignalIceMsg>, target, ConnectionState))
        let signal_channel_map: Arc<
            RwLock<HashMap<String, (UnboundedSender<TunnelSignalIceMsg>, String, ConnectionState)>>,
        > = Arc::new(RwLock::new(HashMap::new()));

        // thread map : tunnel id with future runnable.
        let thread_map: Arc<RwLock<HashMap<String, std::thread::JoinHandle<()>>>> =
            Arc::new(RwLock::new(HashMap::new()));

        // create mpsc IceCommandMsg channel.
        // pass create_tx to TunnelCreatorImp
        let (mut create_tx, mut create_rx) = mpsc::unbounded::<IceCommandMsg>();
        let add_sender = create_tx.clone();

        // sender_tx to send message by websocket
        // pass sender_tx to TunnelCreatorImp.
        let (mut sender_tx, mut sender_rx) = mpsc::unbounded::<(String, TunnelSignalIceMsg)>();

        // create a TunnelCreatorImp, notify service is created ok!
        let creator = TunnelCreatorImp {
            sender_tx: sender_tx.clone(),
            local_client_id: client_id.to_owned(),
            commander: add_sender,
            signal_channel_map: signal_channel_map.clone(),
        };

        // set a rwlock running flag
        let running = Arc::new(RwLock::new(true));

        // create a TunnelConnectMsg channel
        // send to async_new_tunnel_connect,if callback_
        let (tunnel_creator_callback_tx, mut tunnel_creator_callback_rx) =
            mpsc::unbounded::<TunnelConnectMsg>();

        // callback.tunnel creator on ready(creator);
        callback_sender
            .send(TunnelConnectMsg::OnTonicServerCreated(creator))
            .await?;

        // copy service TunnelConnectMsg callback for send
        let mut send_callback_sender = callback_sender.clone();
        // copy service TunnelConnectMsg callback for receive
        let mut receive_callback_sender = callback_sender.clone();

        let mut signal_create_tx = create_tx.clone();
        let signal_receiver_handler = async {
            // wait for signal_receiver
            while let Some(msg) = signal_receiver.next().await {
                log::info!("signal_receiver_handler onMessage: {:?}", &msg);
                match msg {
                    // received WsReceivedMessage::text
                    WsReceivedMessage::Text(msg) => {
                        if let Ok(MessageToReceived { event, data, .. }) =
                            WsReceivedMessage::txt_to_message_to_received(&msg)
                        {
                            if event != TUNNEL_EVENT {
                                log::warn!("signal_receiver_handler event not equ: {:?}", &event);
                                continue;
                            }
                            let mut remote_is_controlling = false;

                            if let Ok(msg) = serde_json::from_str::<TunnelSignalIceMsg>(&data) {
                                let (tunnel_id, device_id) = match &msg {
                                    TunnelSignalIceMsg::RemoteAuth(IceAuth {
                                        tunnel_id,
                                        auth_local_client_id: remote_device_id,
                                        auth_remote_client_id,
                                        is_controlling,
                                        ..
                                    }) => {
                                        log::info!("signal_receiver_handler received RemoteAuth,auth_local_client_id:remote_device_id={} auth_remote_client_id ={}",remote_device_id,auth_remote_client_id);
                                        remote_is_controlling = *is_controlling;
                                        (tunnel_id, remote_device_id)
                                    }
                                    TunnelSignalIceMsg::RemoteCandidate(IceCandidate {
                                        tunnel_id,
                                        can_local_client_id: remote_device_id,
                                        can_remote_client_id,
                                        is_controlling,
                                        ..
                                    }) => {
                                        log::info!("signal_receiver_handler received RemoteCandidate,can_local_client_id:remote_device_id={} can_remote_client_id ={}",remote_device_id,can_remote_client_id);

                                        remote_is_controlling = *is_controlling;
                                        (tunnel_id, remote_device_id)
                                    }
                                    TunnelSignalIceMsg::GatherCandidates {
                                        tunnel_id,
                                        gc_local_client_id: remote_device_id,
                                        gc_remote_client_id,
                                        ..
                                    } => {
                                        log::info!("signal_receiver_handler received GatherCandidates,gc_local_client_id:remote_device_id={} gc_remote_client_id ={}",remote_device_id,gc_remote_client_id);

                                        (tunnel_id, remote_device_id)
                                    }
                                    TunnelSignalIceMsg::Stop {
                                        tunnel_id,
                                        st_local_client_id: _,
                                        st_remote_client_id: _,
                                        kind: _,
                                    } => {
                                        log::warn!(
                                            "signal_receiver_handler Stop tunnel_id: {}",
                                            tunnel_id,
                                        );
                                        // callback from signal_map.
                                        if let Some((receive_tx, _remote_device_id, _)) =
                                            signal_channel_map.write().await.get_mut(tunnel_id)
                                        {
                                            if let Err(e) = receive_tx.send(msg).await {
                                                log::error!(
                                                    "TunnelSignalIceMsg::Stop receive_tx send err: {}",
                                                    &e
                                                );
                                            } else {
                                                log::info!("TunnelSignalIceMsg::Stop send");
                                            }
                                        }
                                        continue;
                                    }
                                };
                                log::info!(
                                    "signal_receiver_handler tunnel_id: {} device_id: {}",
                                    tunnel_id,
                                    device_id
                                );

                                let (receive_tx, receive_rx) =
                                    mpsc::unbounded::<TunnelSignalIceMsg>();
                                let mut need_new = false;

                                let (receive_tx, _, _) = {
                                    let mut map = signal_channel_map.write().await;

                                    // received remote_is_controlling is false from signal. ignore this message
                                    // if remote_is_controlling == false or device_id is null and map not contains key
                                    if (!remote_is_controlling || device_id == "")
                                        && !map.contains_key(tunnel_id)
                                    {
                                        log::warn!("signal_channel_map not contain tunnel_id: {}, but remote_is_controlling: {} or device_id {}", tunnel_id, remote_is_controlling, device_id);
                                        continue;
                                    }

                                    // then insert to map.
                                    map.entry(tunnel_id.to_owned())
                                        .or_insert_with(|| {
                                            need_new = true;
                                            (
                                                receive_tx,
                                                device_id.to_owned(),
                                                ConnectionState::Unspecified,
                                            )
                                        })
                                        .to_owned()
                                };

                                // remote is controlling! need new
                                if need_new {
                                    let tunnel_id = tunnel_id.to_owned();
                                    let remote_device_id = device_id.to_owned();
                                    log::info!("signal_receiver_handler create new");
                                    if let Err(e) = signal_create_tx
                                        .send(IceCommandMsg::Create(IceCreateMsg {
                                            is_controlling: false,
                                            tunnel_id,
                                            remote_client_id: remote_device_id,
                                            receive_rx,
                                        }))
                                        .await
                                    {
                                        log::error!("start_tunnel_service signal_receiver_handler create_tx send err: {}", &e);
                                    }
                                    log::info!("signal_receiver_handler create_tx send");
                                }
                                if let Err(e) = receive_tx.unbounded_send(msg) {
                                    log::error!("start_tunnel_service signal_receiver_handler receive_tx unbounded_send err: {}", &e);
                                } else {
                                    log::info!("signal_receiver_handler receive_tx unbounded_send");
                                }
                            }
                        }
                    }
                    WsReceivedMessage::AckSendResult(WsAckSendResult {
                        is_send_ok,
                        send_msg_id,
                        reason,
                    }) => {
                        // let MessageToReceived {
                        //     target_id: _,
                        //     id: _,
                        //     event,
                        //     data,
                        // } = context;
                        let event_data_opt = signal_adapter.remove_send_msg(send_msg_id.to_owned());
                        if event_data_opt.is_none() {
                            log::warn!(" signal_adapter.remove_send_msg is none:{}", send_msg_id);
                            continue;
                        }
                        if let Some((event, data)) = event_data_opt {
                            if event != TUNNEL_EVENT {
                                log::warn!("signal_receiver_handler event not equ: {:?}", &event);
                                continue;
                            }
                            if let Ok((tunnel_id, device_id)) = get_tunnel_id_from_data(&data) {
                                log::debug!(
                                    "signal_receiver_handler SendResult device_id:{} tunnel_id: {} is_send_ok:{} reason = {} ",
                                    device_id,
                                    tunnel_id,
                                    is_send_ok,
                                    reason
                                );

                                if is_send_ok {
                                    continue;
                                }

                                let kind = TunnelStopKind::SignalSendFail;

                                if let Some((receive_tx, _remote_device_id, state)) =
                                    signal_channel_map.write().await.get_mut(&tunnel_id)
                                {
                                    if_no_kcp_set_err(
                                        &tunnel_id,
                                        anyhow::anyhow!(
                                            "if_no_kcp_set_err send signal failed: {}",
                                            kind.to_string()
                                        ),
                                    )
                                    .await;

                                    if state != &ConnectionState::Connected {
                                        let tm = time_utils::timestamp_ms_i64();
                                        let log_msg = format!(
                                            "signal send state changed to:{:?} failed,tm:{:?}",
                                            device_id.clone(),
                                            tm
                                        );

                                        if let Err(e) = receive_callback_sender
                                            .send(TunnelConnectMsg::OnTunnelStateChange(
                                                TunnelStatus {
                                                    sa_local_client_id: client_id.to_owned(),
                                                    sa_remote_client_id: device_id.to_owned(),
                                                    tunnel_id: tunnel_id.to_owned(),
                                                    msg: log_msg.to_owned(),
                                                    state: ConnectionState::Failed,
                                                },
                                            ))
                                            .await
                                        {
                                            log::error!("receive_callback_sender send err: {}", &e);
                                        }
                                        if let Err(e) = receive_tx
                                            .send(TunnelSignalIceMsg::Stop {
                                                tunnel_id,
                                                st_local_client_id: client_id.to_owned(),
                                                st_remote_client_id: device_id.to_owned(),
                                                kind,
                                            })
                                            .await
                                        {
                                            log::error!(
                                                "TunnelSignalIceMsg::Stop receive_tx send err: {}",
                                                &e
                                            );
                                        } else {
                                            log::info!("SendResult TunnelSignalIceMsg::Stop send");
                                        }
                                    }
                                }
                            }
                        }
                    }
                    _ => {
                        log::debug!("signal_receiver_handler new msg: {:?}", &msg);
                    }
                }
            }
            *running.write().await = false;
            log::info!("signal_receiver_handler end");
        };

        let mut sender_tx_command = sender_tx.clone();

        let command_handler = async {
            while let Some(msg) = create_rx.next().await {
                log::info!("command_handler new msg: {:?}", &msg);
                match msg {
                    IceCommandMsg::DisconnectTunnel(tunnel_id) => {
                        if let Err(e) = disconnect(
                            &signal_channel_map,
                            &mut sender_tx_command,
                            client_id,
                            &tunnel_id,
                        )
                        .await
                        {
                            log::info!(
                                "command_handler DisconnectTunnel tunnel_id:{} err:{:?} ",
                                tunnel_id,
                                &e
                            );
                        }
                    }
                    IceCommandMsg::Create(IceCreateMsg {
                        is_controlling,
                        tunnel_id,
                        remote_client_id,
                        receive_rx,
                    }) => {
                        log::info!("command_handler Create is_controlling:{} tunnel_id:{} remote_device_id: {}", is_controlling, tunnel_id, remote_client_id);
                        let my_device_id = client_id.to_owned();
                        let sender_tx = sender_tx_command.clone();
                        let mut callback_tx = tunnel_creator_callback_tx.clone();

                        let options = options.clone();
                        let tunnel_id_key = tunnel_id.to_owned();
                        let thread_map_move = thread_map.to_owned();

                        let runner = std::thread::spawn(move || {
                            let client_id = my_device_id.to_owned();
                            if let Ok(rt) = Builder::new_multi_thread()
                                .on_thread_start(|| {
                                    // call_attach_fn();
                                })
                                .on_thread_stop(|| {
                                    // call_detach_fn();
                                })
                                .enable_all()
                                .build()
                            {
                                let _res = rt.block_on(
                                    async move {
                                    let mut tunnel_err: Option::<anyhow::Error> = None;
                                    if let Err(e) = TunnelCreatorImp::async_new_tunnel_connect(
                                        is_controlling,
                                        &client_id,
                                        &remote_client_id,
                                        &tunnel_id,
                                        receive_rx,
                                        sender_tx,
                                        callback_tx.clone(),
                                        &options,
                                    )
                                    .await {
                                        log::error!("TunnelCreatorImp async_new_tunnel_connect return err: {}", &e);
                                        tunnel_err = Some(e);
                                    }
                                    let tunnel_err_msg = if let Some(error) = tunnel_err {
                                        error.to_string()
                                    } else {
                                        r#"tunnel connect end"#.to_string()
                                    };
                                    if let Err(e) = callback_tx
                                        .send(TunnelConnectMsg::OnTunnelStateChange(TunnelStatus {
                                            tunnel_id: tunnel_id.to_owned(),
                                            sa_local_client_id: client_id.to_owned(),
                                            sa_remote_client_id: remote_client_id.to_owned(),
                                            msg: tunnel_err_msg.to_owned(),
                                            state: ConnectionState::Closed,
                                        }))
                                        .await
                                    {
                                        log::error!("do_connect end send OnStateChange err: {}", &e);
                                    }
                                    thread_map_move.write().await.remove(&tunnel_id);
                                });
                            }
                        });
                        thread_map.write().await.insert(tunnel_id_key, runner);
                    }
                    IceCommandMsg::GatherCandidates => {
                        log::info!("IceCommandMsg::GatherCandidates");
                        for (key, (tx, device_id, state)) in
                            signal_channel_map.write().await.iter_mut()
                        {
                            tracing::info!(%state, %device_id, "IceCommandMsg::GatherCandidates");
                            if state != &ConnectionState::Disconnected {
                                if state == &ConnectionState::Connected
                                    || state == &ConnectionState::Checking
                                {
                                    *state = ConnectionState::Unspecified; // 设置为自定义的restart标志
                                    tracing::info!(%state, %device_id, "IceCommandMsg::GatherCandidates set Unspecified");
                                }
                                continue;
                            }
                            if let Err(e) = sender_tx_command
                                .send((
                                    device_id.to_owned(),
                                    TunnelSignalIceMsg::GatherCandidates {
                                        tunnel_id: key.to_owned(),
                                        gc_local_client_id: client_id.to_owned(),
                                        gc_remote_client_id: device_id.to_owned(),
                                    },
                                ))
                                .await
                            {
                                log::error!("IceCommandMsg::GatherCandidates sender_tx_command send err: {}", &e);
                            }

                            if let Err(e) = tx
                                .send(TunnelSignalIceMsg::GatherCandidates {
                                    tunnel_id: key.to_owned(),
                                    gc_local_client_id: client_id.to_owned(),
                                    gc_remote_client_id: device_id.to_owned(),
                                })
                                .await
                            {
                                log::error!(
                                    "IceCommandMsg::GatherCandidates tx send to {} err: {}",
                                    key,
                                    &e
                                );
                            }
                        }
                    }
                    IceCommandMsg::StopService => {
                        let map = { signal_channel_map.read().await.clone() };
                        for (tunnel_id, (mut sender, remote_device_id, _)) in map {
                            if let Err(e) = sender_tx_command
                                .send((
                                    remote_device_id.clone(),
                                    TunnelSignalIceMsg::Stop {
                                        tunnel_id: tunnel_id.to_owned(),
                                        st_local_client_id: client_id.to_owned(),
                                        st_remote_client_id: remote_device_id.to_owned(),
                                        kind: TunnelStopKind::RemoteClose,
                                    },
                                ))
                                .await
                            {
                                log::error!("StopService disconnect  send err: {}", &e);
                            }

                            if let Err(e) = sender
                                .send(TunnelSignalIceMsg::Stop {
                                    tunnel_id: tunnel_id.to_owned(),
                                    st_local_client_id: client_id.to_owned(),
                                    st_remote_client_id: remote_device_id.to_owned(),
                                    kind: TunnelStopKind::LocalClose,
                                })
                                .await
                            {
                                log::error!("StopService sender send err: {}", &e);
                            }
                        }
                        tokio::time::sleep(Duration::from_millis(50)).await;
                        let _map: HashMap<
                            String,
                            (UnboundedSender<TunnelSignalIceMsg>, String, ConnectionState),
                        > = {
                            let mut map = signal_channel_map.write().await;
                            map.drain_filter(|_, _| true).collect()
                        };
                        *running.write().await = false;
                        create_tx.close_channel();
                        break;
                    }
                }
            }
            *running.write().await = false;
            log::info!("command_handler end");
        };

        let send_callback_handler = async {
            while *running.read().await {
                match future::select(tunnel_creator_callback_rx.next(), sender_rx.next()).await {
                    Either::Left((msg, _)) => {
                        log::info!("start_tunnel_service callback_sender msg: {:?}", &msg,);
                        if let Some(msg) = msg {
                            let send_destroy = match &msg {
                                TunnelConnectMsg::OnTunnelStateChange(msg) => {
                                    match signal_channel_map.write().await.get_mut(&msg.tunnel_id) {
                                        Some((tx, device_id, state)) => {
                                            if state == &ConnectionState::Unspecified
                                                && msg.state == ConnectionState::Disconnected
                                            {
                                                // if current connection state changged to unspecified and disconnect,
                                                // then send a gather candidate to remote_client_id by websocket.
                                                log::info!("start_tunnel_service send_callback_handler GatherCandidates");
                                                if let Err(e) = sender_tx
                                                    .send((
                                                        device_id.to_owned(),
                                                        TunnelSignalIceMsg::GatherCandidates {
                                                            tunnel_id: msg.tunnel_id.to_owned(),
                                                            gc_local_client_id: client_id
                                                                .to_owned(),
                                                            gc_remote_client_id: device_id
                                                                .to_owned(),
                                                        },
                                                    ))
                                                    .await
                                                {
                                                    log::error!("start_tunnel_service send_callback_handler sender_tx send err: {}", &e);
                                                }

                                                // and notify agent to gather candidate.
                                                if let Err(e) = tx
                                                    .send(TunnelSignalIceMsg::GatherCandidates {
                                                        tunnel_id: msg.tunnel_id.to_owned(),
                                                        gc_local_client_id: client_id.to_owned(),
                                                        gc_remote_client_id: device_id.to_owned(),
                                                    })
                                                    .await
                                                {
                                                    log::error!("start_tunnel_service send_callback_handler tx send to {} err: {}", &msg.tunnel_id, &e);
                                                }
                                            }
                                            *state = msg.state;
                                        }
                                        None => {
                                            log::info!(
                                                "signal_channel_map get none {}",
                                                &msg.tunnel_id
                                            );
                                        }
                                    }

                                    // when state changed to closed or failed
                                    if msg.state == ConnectionState::Closed
                                        || msg.state == ConnectionState::Failed
                                    {
                                        // need clean-up
                                        Some((
                                            msg.tunnel_id.to_owned(),
                                            msg.sa_remote_client_id.to_owned(),
                                        ))
                                    } else {
                                        None
                                    }
                                }
                                _ => None,
                            };

                            // then callback message.
                            if let Err(e) = send_callback_sender.send(msg).await {
                                log::error!(
                                    "start_tunnel_service callback_handler callback_sender send err: {}",
                                    &e
                                );
                            }

                            // need clean up.
                            if let Some((tunnel_id, remote_device_id)) = send_destroy {
                                log::info!("send_destroy tunnel_id: {}", tunnel_id);
                                if let Some((mut receive_tx, _device_id, _)) =
                                    signal_channel_map.write().await.remove(&tunnel_id)
                                {
                                    // send stop message to remote client id
                                    if let Err(e) = receive_tx
                                        .send(TunnelSignalIceMsg::Stop {
                                            tunnel_id: tunnel_id.to_owned(),
                                            st_local_client_id: client_id.to_owned(),
                                            st_remote_client_id: remote_device_id.to_owned(),
                                            kind: TunnelStopKind::Failed,
                                        })
                                        .await
                                    {
                                        log::error!("send_destroy receive_tx send err: {}", &e);
                                    } else {
                                        log::info!("send_destroy send");
                                    }
                                }
                                // callback ontunnel connect destroy.
                                if let Err(e) = send_callback_sender
                                    .send(TunnelConnectMsg::OnTunnelConnectDestroy {
                                        tunnel_id,
                                        local_client_id: client_id.to_owned(),
                                        remote_client_id: remote_device_id.to_owned(),
                                    })
                                    .await
                                {
                                    log::error!(
                                        "start_tunnel_service callback_handler callback_sender send_destroy err: {}",
                                        &e
                                    );
                                }
                            } else {
                                log::info!("send_destroy none");
                            }
                        } else {
                            log::info!("callback_tx has gone break");
                            break;
                        }
                    }
                    Either::Right((msg, _)) => match msg {
                        // send with websocket
                        Some((target_id, msg)) => match serde_json::to_string(&msg) {
                            Ok(data) => {
                                log::info!(
                                    "start_tunnel_service signal send target_id:{} data: {}",
                                    &target_id,
                                    &data,
                                );
                                if let Err(e) = signal_adapter_clone
                                    .send(target_id.to_owned(), TUNNEL_EVENT.to_owned(), data)
                                    .await
                                {
                                    let (tunnel_id, remote_device_id) = get_tunnel_id_from_msg(msg);

                                    let need_stop = {
                                        match signal_channel_map.read().await.get(&tunnel_id) {
                                            Some((_, _, state)) => {
                                                state != &ConnectionState::Connected
                                            }
                                            None => true,
                                        }
                                    };
                                    if need_stop {
                                        let tm = time_utils::timestamp_ms_i64();
                                        let log_msg = format!(
                                            "signal send need_stop to:{:?} failed,tm:{:?}",
                                            remote_device_id.clone(),
                                            tm
                                        );
                                        if let Err(e) = send_callback_sender
                                            .send(TunnelConnectMsg::OnTunnelStateChange(
                                                TunnelStatus {
                                                    tunnel_id,
                                                    sa_local_client_id: client_id.to_owned(),
                                                    sa_remote_client_id: remote_device_id
                                                        .to_owned(),
                                                    state: ConnectionState::Failed,
                                                    msg: log_msg.to_owned(),
                                                },
                                            ))
                                            .await
                                        {
                                            log::error!(
                                                    "start_tunnel_service send_callback_handler callback_sender send err: {}",
                                                    &e
                                                );
                                        }
                                    }

                                    log::error!(
                                            "start_tunnel_service callback_handler sender_rx send err: {}",
                                            &e
                                        );
                                }
                            }
                            Err(e) => {
                                log::error!(
                                            "start_tunnel_service callback_handler serde_json::to_string msg err: {}",
                                            &e
                                        );
                            }
                        },
                        None => {
                            log::info!("sender_tx has gone break");
                            break;
                        }
                    },
                }
            }
            log::info!("send_callback_handler end");
        };
        pin_mut!(
            signal_receiver_handler,
            command_handler,
            send_callback_handler
        );
        future::select(
            send_callback_handler,
            future::select(signal_receiver_handler, command_handler),
        )
        .await;

        if let Err(e) = callback_sender
            .send(TunnelConnectMsg::OnTonicServerDroped(
                TonicServerStopKind::ErrorClose,
            ))
            .await
        {
            log::error!("callback_sender.send err: {:?}", e);
        }
        log::info!("start_tunnel_service end");
        Ok(())
    }
}

pub async fn handle_selected_contents_duplex(
    start: SendContentStart,
    client: &mut TunnelServiceClient<InterceptedService<Channel, FlowControlInterceptor>>,
    outbound_user_callback_tx: tokio::sync::mpsc::UnboundedSender<send_content_msg::Msg>,
) -> Result<(), anyhow::Error> {
    log::info!("handle_selected_as_content_stream start:{:?}", start);

    let (outbound_remote_tx, mut outbound_remote_rx) =
        tokio::sync::mpsc::unbounded_channel::<SendContentMsg>();
    let (inbound_remote_tx, inbound_remote_rx) = tokio::sync::mpsc::unbounded_channel();

    // outbound message
    let outbound = async_stream::stream! {
        while let Some(item) = outbound_remote_rx.recv().await {
            yield item;
        }
    };
    let request = Request::new(outbound);
    let inbound_remote_handler = async {
        //client send request.
        let mut inbound_remote = client.send_content(request).await?.into_inner();
        while let Some(msg) = inbound_remote.message().await? {
            inbound_remote_tx.send(msg)?
        }

        let res: Result<(), anyhow::Error> = Ok(());
        res
    };

    let start_content_exchange_handler = select_content_tunnel_duplex(
        Some(start),
        inbound_remote_rx,
        outbound_remote_tx,
        outbound_user_callback_tx.clone(),
    );
    pin_mut!(inbound_remote_handler, start_content_exchange_handler);

    let res =
        match futures_util::future::select(inbound_remote_handler, start_content_exchange_handler)
            .await
        {
            Either::Left((value1, _)) => value1,
            Either::Right((value2, _)) => value2,
        };
    log::info!("handle_selected_as_content_stream end res:{:?}", res);
    res
}

// start content service with tunnel id
pub async fn select_content_tunnel_duplex(
    start: Option<SendContentStart>,
    mut inbound_remote_rx: tokio::sync::mpsc::UnboundedReceiver<SendContentMsg>,
    outbound_remote_tx: tokio::sync::mpsc::UnboundedSender<SendContentMsg>,
    outbound_user_callback_tx: tokio::sync::mpsc::UnboundedSender<send_content_msg::Msg>,
) -> Result<(), anyhow::Error> {
    log::info!("select_content_tunnel_duplex {:?}", start);
    // key = tunnel id  value = (instant ,send content response)
    let mut user_send_map: HashMap<String, (Instant, SendContentRes)> = HashMap::new();
    // start a channel.
    let (inbound_user_send_tx, mut inbound_user_send_rx) = futures_channel::mpsc::unbounded();

    // async function for loop handler
    // loop handler result.
    let loop_handler = async {
        // if there are SendContentStart
        let mut start_is_controlling = true;
        let mut tunnel_id = if let Some(start) = start {
            let tunnel_id = start.tunnel_id.clone();
            inbound_user_send_tx
                .unbounded_send(SendContentCommand::Start(start))?;
                tunnel_id
        } else {
            "".to_owned()
        };
        if tunnel_id == "" {
            start_is_controlling = false;
        }
        // content ping timeout
        let mut inbound_user_send_tx = Some(inbound_user_send_tx);

        let mut from_user;
        let send_timeout = std::time::Duration::from_secs(5);
        let mut interval = tokio::time::interval(std::time::Duration::from_millis(500));
        let mut last_ping_instant = Instant::now();
        // 15 seconds timeout
        let ping_timeout = std::time::Duration::from_secs(15);
        let mut tick_times = 0;
        let mut last_cpu_run_instant = Instant::now();

        let mut no_drop_contents = VecDeque::<SendContentData>::with_capacity(30);
        loop {
            // select! wait for branch finish.
            let (start, stop, remote, user) = tokio::select! {
                // result = future run end => { result... }
                _ = interval.tick() => {
                    // 500 milliseconds tick()
                    from_user = true;
                    let now = Instant::now();
                    let timeout_map = user_send_map.drain_filter( |_, (timeout, _)| {
                        let is_timeout = &now > timeout;
                        is_timeout
                    });

                    for (_, (_, res)) in timeout_map.into_iter() {
                        outbound_user_callback_tx.send(Msg::Res(res))?;
                    }

                    // 5 seconds send ping
                    tick_times = tick_times + 1;
                    let remote = if tick_times % 10 == 0 {
                        tick_times = 0;
                        Some(Msg::Ping(SendContentPing{}))
                    } else {
                        if tick_times % 4 == 0 {
                            if let Some(no_drop_data) = no_drop_contents.pop_front() {
                                Some(Msg::Data(no_drop_data))
                            }else{
                                None
                            }
                        } else {
                            None
                        }
                    };

                    let stop = if now - last_ping_instant > ping_timeout {
                        // reset last_ping_instant
                        last_ping_instant = now;
                        log::error!("select_content_tunnel_duplex check stale message!");
                        Some(SendContentStop{tunnel_id: tunnel_id.clone(), is_confirm: true})
                    } else {
                        None
                    };
                    (None, stop, remote, None)
                },

                // read data from inbound_user_send_rx
                data = inbound_user_send_rx.next() => {
                    from_user = true;
                    let data = data.ok_or(anyhow::anyhow!("inbound_user_tx closed"))?;
                    match data {
                        SendContentCommand::SendContent(data) => {
                            let SendContentData {id, tunnel_id, my_device_id, remote_device_id, can_drop, ..} = &data;
                            if is_steam_backpressure_heavy(tunnel_id.clone()) {
                                log::info!("flow control content... steam_backpressure_heavy !");
                                let res = SendContentRes {id: id.clone(), tunnel_id: tunnel_id.clone(), my_device_id: my_device_id.clone(), remote_device_id: remote_device_id.clone(), code: -2, msg: "Timeout".to_owned()};
                                let candrop = if let Some(can_drop ) = can_drop {
                                    *can_drop
                                }else{
                                    true
                                };
                                if candrop {
                                    log::warn!("flow control drop content: {:?}",id);
                                    user_send_map.insert(id.clone(), (Instant::now(), res));
                                    (None, None, None, None)
                                }else{
                                    user_send_map.insert(id.clone(), (Instant::now() + send_timeout, res));
                                    // 在block的条件下，推到queue里面，避免出现问题
                                    if no_drop_contents.len() <= no_drop_contents.capacity(){
                                        no_drop_contents.push_back(data);
                                    }else{
                                        // keep 30 count content.
                                        no_drop_contents.pop_front();
                                        no_drop_contents.push_back(data);
                                    }
                                    (None, None, None, None)
                                }
                            }else{
                                while let Some(remote) = no_drop_contents.pop_front() {
                                    outbound_remote_tx.send(SendContentMsg { msg: Some(Msg::Data(remote))})?;
                                    tokio::time::sleep(Duration::from_millis(50)).await;
                                }

                                let res = SendContentRes {id: id.clone(), tunnel_id: tunnel_id.clone(), my_device_id: my_device_id.clone(), remote_device_id: remote_device_id.clone(), code: -2, msg: "Timeout".to_owned()};
                                user_send_map.insert(id.clone(), (Instant::now() + send_timeout, res));
                                (None, None, Some(Msg::Data(data)), None)
                            }
                        },
                        SendContentCommand::Stop(stop) => (None, Some(stop.clone()), Some(Msg::Stop(stop.clone())), None),
                        SendContentCommand::Start(start) => (Some(start.clone()), None, Some(Msg::Start(start.clone())), Some(Msg::Start(start))),
                    }
                },

                // read message from inbound_remote_rx
                msg = inbound_remote_rx.recv() => {
                    from_user = false;
                    let msg = msg.ok_or(anyhow::anyhow!("inbound_remote finished"))?;
                    let SendContentMsg {msg} = msg;
                    last_ping_instant = Instant::now();
                    if let Some(msg) = msg {
                        let (start, stop, remote, user) = match msg {
                            Msg::Ping(_) => {
                                (None, None, Some(Msg::Pong(SendContentPong{})), None)
                            }
                            Msg::Pong(_) => {
                                (None, None, None, None)
                            }
                            Msg::Data(mut data) => {
                                let user_data = data.clone();
                                let id = data.id.to_owned();
                                let tunnel_id = data.tunnel_id.to_owned();
                                let my_device_id = data.my_device_id.to_owned();
                                let remote_device_id = data.remote_device_id.to_owned();
                                mem::swap(&mut data.my_device_id, &mut data.remote_device_id);
                                let (code, msg) = if outbound_user_callback_tx.is_closed() {
                                    (-1, "outbound_user_tx closed".to_owned())
                                } else {
                                    (0, "OK". to_owned())
                                };
                                (None, None, Some(Msg::Res(SendContentRes {id, tunnel_id, my_device_id, remote_device_id, code, msg })), Some(Msg::Data(user_data)))
                            }
                            Msg::Res(res) => {
                                let user = match user_send_map.remove(&res.id) {
                                    None => None,
                                    Some(_) => Some(Msg::Res(res)),
                                };
                                (None, None, None, user)
                            }
                            Msg::Start(mut start) => {
                                if !start_is_controlling{
                                    log::info!("select_content_tunnel_duplex start exchagne local_client_id and remote_client_id");
                                    let local_client_id = start.remote_device_id.to_owned();
                                    let remote_client_id = start.my_device_id.to_owned();

                                    start.my_device_id = local_client_id.to_owned();
                                    start.remote_device_id = remote_client_id.to_owned();
                                }
                                tunnel_id = start.tunnel_id.clone();
                                (Some(start.clone()), None, None,  Some(Msg::Start(start)))
                            }
                            Msg::Stop(mut stop) => {
                                let remote = if stop.is_confirm {
                                    None
                                } else {
                                    stop.is_confirm = true;
                                    Some(Msg::Stop(stop.clone()))
                                };
                                (None, Some(stop.clone()), remote, None)
                            }
                        };
                        (start, stop, remote, user)
                    } else {
                        log::warn!("SendContentMsg msg none is possible?");
                        continue;
                    }
                }
            };

            /////////////// FIX: cpu run into deep sleep and back, need check ping_timeout.///////////////////
            let _now = Instant::now();
            let need_stop = if _now - last_cpu_run_instant > ping_timeout {
                true
            } else {
                false
            };
            if need_stop {
                log::warn!("select_content_tunnel_duplex stale need_stop because now is so far behind last_cpu_run_instant, may be cpu sleep so long...");
                if let Err(err) = close_remote_tonic_tunnel(tunnel_id){
                    log::error!("select_content_tunnel_duplex need_stop close_remote_tonic_tunnel error:{}",err);
                }
                break;
            }
            last_cpu_run_instant = _now;
            //////////////////////////////////////////////////////////////////////////////////

            if let Some(start) = start {
                log::info!("select_content_tunnel_duplex start {:?}", start);
                let SendContentStart {
                    tunnel_id,
                    my_device_id,
                    remote_device_id,
                    is_controlling:_,
                } = start;
                if let Some(inbound_user_send_tx) = inbound_user_send_tx.take() {
                    get_or_set_content_sender(
                        true,
                        &tunnel_id,
                        Some((inbound_user_send_tx, my_device_id, remote_device_id)),
                    )?;
                }
            }

            if let Some(remote) = remote {
                // log::debug!("start_content_exchange remote {:?}", remote);
                outbound_remote_tx.send(SendContentMsg { msg: Some(remote) })?;
            }

            if let Some(user) = user {
                log::debug!("select_content_tunnel_duplex user {:?} outbound_user_callback_tx is_closed = {:?}", user,outbound_user_callback_tx.is_closed());
                outbound_user_callback_tx.send(user)?;
            }

            if let Some(stop) = stop {
                log::warn!("select_content_tunnel_duplex stop {:?} from_user: {}", stop, from_user);
                if stop.is_confirm {
                    log::info!("select_content_tunnel_duplex stop is_confirm true close");
                    if from_user {
                        tokio::time::sleep(std::time::Duration::from_millis(500)).await; //wait for send stop to remote
                    } else {
                        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
                    }
                    if let Err(err) = close_remote_tonic_tunnel(tunnel_id){
                        log::error!("select_content_tunnel_duplex need_stop close_remote_tonic_tunnel error:{}",err);
                    }
                    break;
                } else {
                    log::info!("select_content_tunnel_duplex stop is_confirm false");
                }
            }
        }
        let res: Result<(), anyhow::Error> = Ok(());
        res
    }.await;

    log::info!(
        "select_content_tunnel_duplex loop end res: {:?}",
        loop_handler
    );
    for (_, (_, mut res)) in user_send_map.into_iter() {
        res.code = -3;
        res.msg = "Stoped".to_owned();
        outbound_user_callback_tx.send(Msg::Res(res))?;
    }

    while let Ok(msg) = inbound_user_send_rx.try_next() {
        if let Some(msg) = msg {
            match msg {
                SendContentCommand::SendContent(data) => {
                    let SendContentData {
                        id,
                        tunnel_id,
                        my_device_id,
                        remote_device_id,
                        ..
                    } = data;
                    let res = SendContentRes {
                        id,
                        tunnel_id,
                        my_device_id,
                        remote_device_id,
                        code: -3,
                        msg: "Stoped".to_owned(),
                    };
                    outbound_user_callback_tx.send(Msg::Res(res))?;
                }
                _ => {}
            }
        } else {
            break;
        }
    }
    loop_handler?;

    Ok(())
}

pub async fn handle_selected_bytebuffer_duplex(
    start: ByteBufferStart,
    client: &mut TunnelServiceClient<InterceptedService<Channel, FlowControlInterceptor>>,
    outbound_user_callback_tx: tokio::sync::mpsc::UnboundedSender<Bfmsg>,
) -> Result<(), anyhow::Error> {
    log::info!("handle_selected_bytebuffer_duplex start:{:?}", start);

    let (outbound_remote_tx, mut outbound_remote_rx) =
        tokio::sync::mpsc::unbounded_channel::<ByteBufferMsg>();
    let (inbound_remote_tx, inbound_remote_rx) = tokio::sync::mpsc::unbounded_channel();

    // outbound message
    let outbound = async_stream::stream! {
        while let Some(item) = outbound_remote_rx.recv().await {
            yield item;
        }
    };
    log::info!("handle_selected_bytebuffer_duplex outbound");
    let request = Request::new(outbound);
    let inbound_remote_handler = async {
        //client send bytebuffer.
        let mut inbound_remote = client.send_bytebuffer(request).await?.into_inner();
        while let Some(msg) = inbound_remote.message().await? {
            inbound_remote_tx.send(msg)?
        }
        let res: Result<(), anyhow::Error> = Ok(());
        res
    };

    let start_byte_exchange_handler = select_bytebuf_tunnel_duplex(
        Some(start),
        inbound_remote_rx,
        outbound_remote_tx,
        outbound_user_callback_tx.clone(),
    );
    pin_mut!(inbound_remote_handler, start_byte_exchange_handler);

    let res =
        match futures_util::future::select(inbound_remote_handler, start_byte_exchange_handler)
            .await
        {
            Either::Left((value1, _)) => value1,
            Either::Right((value2, _)) => value2,
        };
    log::info!("handle_selected_bytebuffer_duplex end res:{:?}", res);
    res
}

// start bytebuffer service with tunnel id
pub async fn select_bytebuf_tunnel_duplex(
    start: Option<ByteBufferStart>,
    mut inbound_remote_rx: tokio::sync::mpsc::UnboundedReceiver<ByteBufferMsg>,
    outbound_remote_tx: tokio::sync::mpsc::UnboundedSender<ByteBufferMsg>,
    outbound_user_callback_tx: tokio::sync::mpsc::UnboundedSender<byte_buffer_msg::Bfmsg>,
) -> Result<(), anyhow::Error> {
    log::info!("select_bytebuf_tunnel_duplex {:?}", start);
    // key = tunnel id  value = (instant ,send content response)
    let mut user_send_map: HashMap<String, (Instant, ByteBufferRes)> = HashMap::new();
    // start a channel.
    let (inbound_user_send_tx, mut inbound_user_send_rx) = futures_channel::mpsc::unbounded();

    // async function for loop handler
    // loop handler result.
    let loop_handler = async {
        // if there are SendContentStart
        let mut start_is_controlling = true;
        let mut tunnel_id = if let Some(start) = start {
            let tunnel_id = start.tunnel_id.clone();
            inbound_user_send_tx
                .unbounded_send(SendByteBufferCommand::Start(start))?;
                tunnel_id
        } else {
            "".to_owned()
        };
        if tunnel_id == "" {
            start_is_controlling = false;
        }
        // content ping timeout
        let mut inbound_user_send_tx = Some(inbound_user_send_tx);

        let mut from_user;
        let send_timeout = std::time::Duration::from_secs(5);
        let mut interval = tokio::time::interval(std::time::Duration::from_millis(500));
        let mut last_ping_instant = Instant::now();
        // 15 seconds timeout
        let ping_timeout = std::time::Duration::from_secs(15);
        let mut tick_times = 0;
        let mut last_cpu_run_instant = Instant::now();
        let mut iframe_queue = VecDeque::<ByteBufferData>::new();
        loop {
            // select! wait for branch finish.
            let (start, stop, remote, user) = tokio::select! {
                // result = future run end => { result... }
                _ = interval.tick() => {
                    // 500 milliseconds tick()
                    from_user = true;
                    let now = Instant::now();
                    let timeout_map = user_send_map.drain_filter( |_, (timeout, _)| {
                        let is_timeout = &now > timeout;
                        is_timeout
                    });

                    for (_, (_, res)) in timeout_map.into_iter() {
                        outbound_user_callback_tx.send(Bfmsg::BfRes(res))?;
                    }

                    // 5 seconds send ping
                    tick_times = tick_times + 1;
                    let remote = if tick_times % 10 == 0 {
                        tick_times = 0;
                        Some(Bfmsg::BfPing(ByteBufferPing{}))
                    } else {
                        if tick_times % 4 == 0 {
                            if let Some(iframe) = iframe_queue.pop_front() {
                                Some(Bfmsg::BfData(iframe))
                            }else{
                                None
                            }
                        } else {
                            None
                        }
                    };

                    let stop = if now - last_ping_instant > ping_timeout {
                        // reset last_ping_instant
                        last_ping_instant = now;
                        log::error!("select_bytebuf_tunnel_duplex check stale message!");
                        Some(ByteBufferStop{tunnel_id: tunnel_id.clone(), is_confirm: true})
                    } else {
                        None
                    };
                    (None, stop, remote, None)
                },
                // read data from inbound_user_send_rx
                data = inbound_user_send_rx.next() => {
                    from_user = true;
                    let data = data.ok_or(anyhow::anyhow!("inbound_user_tx closed"))?;
                    match data {
                        SendByteBufferCommand::SendByteBuffer(data) => {
                            let ByteBufferData {id, tunnel_id, video_buffer: _,audio_buffer:_,can_drop} = &data;
                            if is_steam_backpressure_heavy(tunnel_id.clone()) {
                                log::info!("flow control buffer... steam_backpressure_heavy !");
                                let res = ByteBufferRes {id: id.clone(), tunnel_id: tunnel_id.clone(), code: -2, msg: "Timeout".to_owned()};
                                let candrop = if let Some(can_drop ) = can_drop {
                                    *can_drop
                                }else{
                                    true
                                };
                                if candrop {
                                    log::warn!("flow control drop bytebuffer: {:?}",id);
                                    user_send_map.insert(id.clone(), (Instant::now(), res));
                                    (None, None, None, None)
                                } else {
                                    log::warn!("flow control replace i-frame: {:?}",id);
                                    user_send_map.insert(id.clone(), (Instant::now() + send_timeout, res));
                                    // 在block的条件下，只发i帧
                                    iframe_queue.pop_front();
                                    iframe_queue.push_back(data.to_owned());

                                    (None, None, None, None)
                                }
                            }else{
                                if let Some(iframe) = iframe_queue.pop_front() {
                                    outbound_remote_tx.send(ByteBufferMsg { bfmsg: Some(Bfmsg::BfData(iframe)) })?;
                                    tokio::time::sleep(Duration::from_millis(50)).await;
                                }
                                let res = ByteBufferRes {id: id.clone(), tunnel_id: tunnel_id.clone(), code: -2, msg: "Timeout".to_owned()};
                                user_send_map.insert(id.clone(), (Instant::now() + send_timeout, res));
                                (None, None, Some(Bfmsg::BfData(data)), None)
                            }
                        },
                        SendByteBufferCommand::Stop(stop) => (None, Some(stop.clone()), Some(Bfmsg::BfStop(stop.clone())), None),
                        SendByteBufferCommand::Start(start) => (Some(start.clone()), None, Some(Bfmsg::BfStart(start.clone())), Some(Bfmsg::BfStart(start))),
                    }
                },

                // read message from inbound_remote_rx
                msg = inbound_remote_rx.recv() => {
                    from_user = false;
                    let msg = msg.ok_or(anyhow::anyhow!("inbound_remote finished"))?;
                    let ByteBufferMsg {bfmsg} = msg;
                    last_ping_instant = Instant::now();
                    if let Some(bfmsg_opt) = bfmsg {
                        let (start, stop, remote, user) = match bfmsg_opt {
                            Bfmsg::BfPing(_) => {
                                (None, None, Some(Bfmsg::BfPong(ByteBufferPong{})), None)
                            },
                            Bfmsg::BfPong(_) => {
                                (None, None, None, None)
                            },
                            Bfmsg::BfData(data) => {
                                let user_data = data.clone();
                                let id = data.id.to_owned();
                                let tunnel_id = data.tunnel_id.to_owned();
                                // let my_device_id = data.my_device_id.to_owned();
                                // let remote_device_id = data.remote_device_id.to_owned();
                                // mem::swap(&mut data.my_device_id, &mut data.remote_device_id);
                                let (code, msg) = if outbound_user_callback_tx.is_closed() {
                                    (-1, "send bytebuffer outbound_user_tx closed".to_owned())
                                } else {
                                    (0, "OK". to_owned())
                                };
                                if is_steam_backpressure_heavy(tunnel_id.clone()) {
                                    log::warn!("is_steam_backpressure_heavy ,ignore BfRes id:{}",id);
                                    (None, None, None, Some(Bfmsg::BfData(user_data)))
                                }else{
                                    (None, None, Some(Bfmsg::BfRes(ByteBufferRes {id, tunnel_id, code, msg })), Some(Bfmsg::BfData(user_data)))
                                }
                            },
                            Bfmsg::BfRes(res) => {
                                let user = match user_send_map.remove(&res.id) {
                                    None => None,
                                    Some(_) => Some(Bfmsg::BfRes(res)),
                                };
                                (None, None, None, user)
                            },
                            Bfmsg::BfStart(mut start) => {
                                if !start_is_controlling{
                                    log::info!("start_bytebuffer_exchange start exchagne local_client_id and remote_client_id");
                                    let local_client_id = start.remote_device_id.to_owned();
                                    let remote_client_id = start.my_device_id.to_owned();

                                    start.my_device_id = local_client_id.to_owned();
                                    start.remote_device_id = remote_client_id.to_owned();
                                }
                                tunnel_id = start.tunnel_id.clone();
                                (Some(start.clone()), None, None,  Some(Bfmsg::BfStart(start)))
                            },
                            Bfmsg::BfStop(mut stop) => {
                                let remote = if stop.is_confirm {
                                    None
                                } else {
                                    stop.is_confirm = true;
                                    Some(Bfmsg::BfStop(stop.clone()))
                                };
                                (None, Some(stop.clone()), remote, None)
                            }
                        };
                        (start, stop, remote, user)
                    } else {
                        log::warn!("select_bytebuf_tunnel_duplex msg none is possible?");
                        continue;
                    }
                }
            };

            /////////////// FIX: cpu run into deep sleep and back, need check ping_timeout.///////////////////
            let _now = Instant::now();
            let need_stop = if _now - last_cpu_run_instant > ping_timeout {
                true
            } else {
                false
            };
            if need_stop {
                log::warn!("select_bytebuf_tunnel_duplex stale need_stop because now is so far behind last_cpu_run_instant, may be cpu sleep so long...");
                if let Err(err) = close_remote_tonic_tunnel(tunnel_id){
                    log::error!("select_bytebuf_tunnel_duplex need_stop close_remote_tonic_tunnel error:{}",err);
                }
                break;
            }
            last_cpu_run_instant = _now;
            //////////////////////////////////////////////////////////////////////////////////

            if let Some(start) = start {
                log::info!("select_bytebuf_tunnel_duplex start {:?}", start);
                let ByteBufferStart {
                    tunnel_id,
                    my_device_id,
                    remote_device_id,
                    is_controlling:_,
                } = start;
                if let Some(inbound_user_send_tx) = inbound_user_send_tx.take() {
                    get_or_set_bytebuffer_sender(
                        true,
                        &tunnel_id,
                        Some((inbound_user_send_tx, my_device_id, remote_device_id)),
                    )?;
                }
            }

            if let Some(remote) = remote {
                // log::debug!("start_bytebuffer_exchange remote {:?}", remote);
                outbound_remote_tx.send(ByteBufferMsg { bfmsg: Some(remote) })?;
            }

            if let Some(user) = user {
                log::debug!("select_bytebuf_tunnel_duplex user = {:?} ,outbound_user_callback_tx isclosed = {:?}", user, outbound_user_callback_tx.is_closed());
                outbound_user_callback_tx.send(user)?;
            }

            if let Some(stop) = stop {
                log::warn!("select_bytebuf_tunnel_duplex stop {:?} from_user: {}", stop, from_user);
                if stop.is_confirm {
                    log::info!("select_bytebuf_tunnel_duplex stop is_confirm true close");
                    if from_user {
                        tokio::time::sleep(std::time::Duration::from_millis(500)).await; //wait for send stop to remote
                    } else {
                        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
                    }
                    if let Err(err) = close_remote_tonic_tunnel(tunnel_id){
                        log::error!("select_bytebuf_tunnel_duplex need_stop close_remote_tonic_tunnel error:{}",err);
                    }
                    break;
                } else {
                    log::info!("select_bytebuf_tunnel_duplex stop is_confirm false");
                }
            }
        }
        let res: Result<(), anyhow::Error> = Ok(());
        res
    }.await;

    log::info!(
        "select_bytebuf_tunnel_duplex loop end res: {:?}",
        loop_handler
    );
    for (_, (_, mut res)) in user_send_map.into_iter() {
        res.code = -3;
        res.msg = "Stoped".to_owned();
        outbound_user_callback_tx.send(Bfmsg::BfRes(res))?;
    }

    while let Ok(msg) = inbound_user_send_rx.try_next() {
        if let Some(msg) = msg {
            match msg {
                SendByteBufferCommand::SendByteBuffer(data) => {
                    let ByteBufferData { id, tunnel_id, .. } = data;
                    let res = ByteBufferRes {
                        id,
                        tunnel_id,
                        code: -3,
                        msg: "Stoped".to_owned(),
                    };
                    outbound_user_callback_tx.send(Bfmsg::BfRes(res))?;
                }
                _ => {}
            }
        } else {
            break;
        }
    }
    loop_handler?;

    Ok(())
}
